package com.example.webflux;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;

import java.util.stream.Stream;

public class WebFluxTest01 {
    public static void main(String[] args) throws InterruptedException {
        Subscriber<String> subscriber=new Subscriber<String>() {

            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription=subscription;
                System.out.println("第一次建立关系时调用");
                subscription.request(1); //第一次建立关系必须先请求一下数据

            }

            @Override
            public void onNext(String s) {
                System.out.println("订阅者得到的数据："+s);
                //todo 业务处理
                subscription.request(1);


            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("处理异常");

            }

            @Override
            public void onComplete() {
                System.out.println("发布者停止发布");

            }
        };
        Flux<String> flux=Flux.fromStream(Stream.of("abcdefghijkmlnopqrstuvwxyz").flatMap(i->Stream.of(i.split(""))));
        flux.subscribe(subscriber);
        Thread.currentThread().join();

    }
}
